package ar.edu.itba.pod.legajo45002;

import java.rmi.NotBoundException;
import java.rmi.RemoteException;
import java.rmi.registry.LocateRegistry;
import java.rmi.registry.Registry;

import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;

import org.joda.time.Duration;

import ar.edu.itba.balance.api.AgentsBalancer;
import ar.edu.itba.balance.api.AgentsTransfer;
import ar.edu.itba.balance.api.NodeAgent;
import ar.edu.itba.balance.api.NotCoordinatorException;
import ar.edu.itba.event.EventInformation;
import ar.edu.itba.event.RemoteEventDispatcher;
import ar.edu.itba.node.Node;
import ar.edu.itba.node.NodeInformation;
import ar.edu.itba.node.api.ClusterAdministration;
import ar.edu.itba.pod.agent.market.Consumer;
import ar.edu.itba.pod.agent.market.Market;
import ar.edu.itba.pod.agent.market.Producer;
import ar.edu.itba.pod.agent.market.Resource;
import ar.edu.itba.pod.agent.runner.Agent;
import ar.edu.itba.pod.agent.runner.Simulation;
import ar.edu.itba.pod.time.TimeMapper;
import ar.edu.itba.pod.time.TimeMappers;

public class DistributedNodeImpl implements Runnable {
	private static Map<String, NodeInformation> nodes = new ConcurrentHashMap<String, NodeInformation>();

	private static Map<NodeInformation, Integer> agentsInNodes = new ConcurrentHashMap<NodeInformation, Integer>();

	//	private static AtomicInteger totalAgentsNumberInNodes = new AtomicInteger();
	//	private static List<NodeAgent> allAgents = new LinkedList<NodeAgent>();

	private static RemoteEventDispatcher remoteDispatcher;
	private static NodeInformation coordinator;
	private static NodeInformation MynodeInfo;
	private static ClusterAdministration cluster;
	private static AgentsBalancer balancer;
	private static AgentsTransfer agentsTransfer;
	private static Registry registry;
	private static Simulation simulation;

	public DistributedNodeImpl(NodeInformation info, TimeMapper timeMapper) {
		MynodeInfo = info;
		coordinator = MynodeInfo;
		//	this.firtNode = firstNode;

		try {
			cluster = new ClusterAdministratorImpl(info);
		} catch (RemoteException e) {
			e.printStackTrace();
		}
		try {
			balancer = new AgentBalancerImpl(info);
		} catch (RemoteException e) {
			e.printStackTrace();
		}
		try {
			remoteDispatcher = new RemoteEventDispatcherImpl();
		} catch (RemoteException e) {
			e.printStackTrace();
		}
		try {
			agentsTransfer = new AgentsTransferImpl();
		} catch (RemoteException e) {
			e.printStackTrace();
		}


		simulation = new LocalSimImpl(timeMapper);

		Thread remoteEventsPoolThread = new Thread(this);
		remoteEventsPoolThread.start();
	}

	@Override
	public void run() {

		while (true) {
			try {
				Thread.sleep(1000);
			} catch (InterruptedException e1) {
				e1.printStackTrace();
			}
			if (!((RemoteEventDispatcherImpl) remoteDispatcher).getPendingEventsQueue().isEmpty()) {
				EventInformation event = ((RemoteEventDispatcherImpl) remoteDispatcher).getEventFromPending();

				if (event != null) {
					try {
						((MyMultithreadEventDispatcher) ((LocalSimImpl) simulation).dispatcher()).myLocalPublish(event.source(),
								event.event());
					} catch (InterruptedException e) {
						e.printStackTrace();
					}
					NodeInformation noud = null;
					try {
						for (NodeInformation aNode : allNodes().values()) {
							noud = aNode;
							if (!aNode.equals(MynodeInfo)) {
								Registry remoteRegistry = LocateRegistry
								.getRegistry(aNode.host(), aNode.port());
								RemoteEventDispatcher stub;
								try {
									stub = (RemoteEventDispatcher) remoteRegistry.lookup(Node.DISTRIBUTED_EVENT_DISPATCHER);
									stub.publish(event);
									
								}catch (RemoteException e) {
									System.out.println("RemoteException, disconecting node: "+aNode.id()+" from cluster ");
									try {
										cluster.disconnectFromGroup(aNode);
									} catch (NotBoundException e1) {									
									}		
								}catch (Exception e) {
									System.out.println("Exception, disconecting node: "+aNode.id()+" from cluster ");
									try {
										cluster.disconnectFromGroup(aNode);
									} catch (NotBoundException e1) {									
									}					
								}
							}
						}
					} catch (RemoteException e) {
						try {
							System.out.println("RemoteException..., disconecting node: "+noud.id()+" from cluster ");
							cluster.disconnectFromGroup(noud);
						} catch (NotBoundException e1) {
							System.out.println(e1.getMessage());
						} catch (RemoteException e1) {
							System.out.println(e1.getMessage());
						}
					}
					((RemoteEventDispatcherImpl) remoteDispatcher).addProcessedEvent(event);
				}
			}
		}
	}


	public static void addAgentToCluster(Agent agent){

		NodeAgent nodeagent = new NodeAgent(MynodeInfo, agent);
		AgentsBalancer stub;

		try {
			Registry reg = LocateRegistry.getRegistry(coordinator.host(),coordinator.port());
			stub = (AgentsBalancer)reg.lookup(Node.AGENTS_BALANCER);
			stub.addAgentToCluster(nodeagent);
		} catch (RemoteException e) {
			try {
				try {
					cluster.disconnectFromGroup(coordinator);
				} catch (NotBoundException e1) {
				}
				balancer.bullyElection(MynodeInfo, System.nanoTime());
			} catch (RemoteException e1) {
				System.out.println(e1.getMessage());
			}
			//.printStackTrace();
		} catch (NotCoordinatorException e) {
			//	e.printStackTrace();
			try {
				balancer.bullyElection(MynodeInfo, System.nanoTime());
			} catch (RemoteException e1) {
				System.out.println(e1.getMessage());
			}
		} catch (NotBoundException e) {
			// TODO Auto-generated catch block
		//	e.printStackTrace();
		}
	}


	public static void balanceAgents() throws RemoteException, NotBoundException {
		System.out.println("balancing agents...");

		List<NodeAgent>nodeAgentsToBalance = new LinkedList<NodeAgent>();
		Registry registry;
		AgentsTransfer stub;
		int numberOfAgents = 0;

		for( NodeInformation aNode: allNodes().values()){
			registry = LocateRegistry.getRegistry(aNode.host(),aNode.port());
			stub = (AgentsTransfer)registry.lookup(Node.AGENTS_TRANSFER);
			numberOfAgents += stub.getNumberOfAgents();
		}
		System.out.println("total agents in cluster: " +numberOfAgents);

		int agentsForEachNode;
		//	if( numberOfAgents %2 == 0){
		agentsForEachNode = numberOfAgents / allNodes().values().size();
		if( agentsForEachNode == 0)
			agentsForEachNode = 1;
		//	}else{
		//		agentsForEachNode = numberOfAgents / allNodes().values().size() + 1;
		//	}

		Map<NodeInformation, Integer> map = new HashMap<NodeInformation, Integer>();

		for( NodeInformation aNode: allNodes().values()){
			registry = LocateRegistry.getRegistry(aNode.host(),aNode.port());
			stub = (AgentsTransfer)registry.lookup(Node.AGENTS_TRANSFER);
			int aNodeNumberOfAgents = stub.getNumberOfAgents();
			if( aNodeNumberOfAgents > agentsForEachNode){
				nodeAgentsToBalance.addAll( stub.stopAndGet(aNodeNumberOfAgents - agentsForEachNode ));
			}else if ( aNodeNumberOfAgents < agentsForEachNode){
				map.put(aNode, aNodeNumberOfAgents);
			}else{

			}
		}

		int agentsCounter = 0;
		int nodeCounter = 0;
		int cantNodes = map.values().size();

		for( NodeInformation n : map.keySet()){
			System.out.println("going to balance node: "+ n.id());
			registry = LocateRegistry.getRegistry(n.host(),n.port());
			stub = (AgentsTransfer)registry.lookup(Node.AGENTS_TRANSFER);
			int agents = map.get(n);
			List<NodeAgent>agentsForOneNode = new LinkedList<NodeAgent>();

			agentsForOneNode.addAll(nodeAgentsToBalance.subList(agentsCounter,agentsForEachNode - agents  ));


			agentsCounter += agentsForEachNode - agents;
			stub.runAgentsOnNode(agentsForOneNode);

			//	nodeCounter++;

		}

		for( NodeInformation n : allNodes().values()){
			if( nodeAgentsToBalance.size() > agentsCounter){
				System.out.println("going to balance node: "+ n.id());
				registry = LocateRegistry.getRegistry(n.host(),n.port());
				stub = (AgentsTransfer)registry.lookup(Node.AGENTS_TRANSFER);
				List<NodeAgent>agentsForOneNode = new LinkedList<NodeAgent>();

				agentsForOneNode.add(nodeAgentsToBalance.get(agentsCounter));
				agentsCounter++;
				stub.runAgentsOnNode(agentsForOneNode);
			}
		}
	}


	public static void reOrganiceAgents(List<NodeAgent> agents) throws RemoteException, NotBoundException {

		NodeInformation nodeToShutDown = agents.get(0).node();
		System.out.println("reorganiceing agents of node: "+nodeToShutDown.id());
		int agentsToShareCounter = 0;

		int nodesCounter = 0;

		int numberOfNodesRemaining = allNodes().values().size() - 1 ;
		System.out.println("my nodes list: "+ allNodes().values());
		int agentsToShare = agents.size() / numberOfNodesRemaining;
		if( agentsToShare == 0){
			agentsToShare = 1;
		}
		for(NodeInformation aNode : allNodes().values()){
			List<NodeAgent> shareNodes = new LinkedList<NodeAgent>();
			if( !aNode.equals(nodeToShutDown)){
				nodesCounter++;
				if (nodesCounter == numberOfNodesRemaining ){
					shareNodes.addAll(agents.subList(agentsToShareCounter, agents.size()));
				}else{
					shareNodes.addAll(agents.subList(agentsToShareCounter, agentsToShareCounter + agentsToShare));
				}
				agentsToShareCounter += agentsToShare;

				Registry registry = LocateRegistry.getRegistry(aNode.host(),aNode.port());
				AgentsTransfer stub = (AgentsTransfer)registry.lookup(Node.AGENTS_TRANSFER);
				System.out.println("sending agents to node: " + aNode.id());
				stub.runAgentsOnNode(shareNodes);

			}
		}
		//	disconnectMe();
	}

	public static void createGroup(){
		try {
			cluster.createGroup();
		} catch (RemoteException e) {
			// TODO Auto-generated catch block
		//	e.printStackTrace();
		}
	}

	public static void connectToGroup(String args, Integer integer) {

		try {
			cluster.connectToGroup(args	, integer);
		} catch (RemoteException e) {
			//e.printStackTrace();
			System.out.println("Failed connecting to node: "+ args );
		} catch (NotBoundException e) {
			e.printStackTrace();
		}

	}

	public static void shutMeDown(){
		System.out.println("Shutting me down gracefully...");

		try {
			((ClusterAdministratorImpl)cluster).shutDownAgentsOfNode(MynodeInfo);
		} catch (RemoteException e) {
			try {
				cluster.disconnectFromGroup(coordinator);
			} catch (RemoteException e1) {
				System.out.println(e1.getMessage());
			} catch (NotBoundException e1) {
			}
			//		e.printStackTrace();
		} catch (NotBoundException e) {
			//	e.printStackTrace();
		}

		try {
			cluster.disconnectFromGroup(MynodeInfo);
		} catch (RemoteException e) {
			System.out.println(e.getMessage());
			//e.printStackTrace();
		} catch (NotBoundException e) {
		//	e.printStackTrace();
		}
	}

	public static void disconnectMe() {
		nodes = new ConcurrentHashMap<String, NodeInformation>();
	}

	public static Simulation getSimulation() {
		return simulation;
	}

	public static RemoteEventDispatcher getRemoteDispatcher() {
		return remoteDispatcher;
	}

	public static Map<String, NodeInformation> allNodes() {
		return nodes;
	}

	public static NodeInformation getCoordinator() {
		return coordinator;
	}

	public static void addNode(NodeInformation info) {
		nodes.put(info.id(), info);
	}

	public static NodeInformation getMynodeInfo() {
		return MynodeInfo;
	}

	public static AgentsBalancer getBalancer() {
		return balancer;
	}

	public static void setRegistry(Registry registry) {
		DistributedNodeImpl.registry = registry;
	}

	public static Registry getRegistry() {
		return registry;
	}

	public static void setCoordinator(NodeInformation node) {
		coordinator = node;
	}
	public static ClusterAdministration getCluster() {
		return cluster;
	}


}
